import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:isolate';
import 'dart:typed_data';

import 'package:get/get.dart';
import 'package:dartssh2/dartssh2.dart';
import 'package:event_bus/event_bus.dart';
import 'package:rcmd/app/utils/sp_utils/sp_utils.dart';
import 'package:rcmd/app/utils/tools/tools.dart';
import 'package:yaml/yaml.dart';

typedef AbortCallback = void Function();

class CmdResult {
  final bool ok;
  final String msg;
  CmdResult(this.ok, this.msg);
}

class SshCfg {
  final String ip;
  final String user;
  final String pswd;
  final int port;
  SshCfg(this.ip, this.user, this.pswd, this.port);
  SshCfg.fromList(List list)
      : this(list[0], list[1], list[2], int.parse(list[3], radix: 10));

  Future<SSHClient?> get client async {
    var sock = await SSHSocket.connect(this.ip, this.port,
        timeout: Duration(seconds: 2));
    return SSHClient(sock, username: this.user, onPasswordRequest: () => this.pswd);
  }

  Future<SSHSession?> get session async {
    final c = await client;
    final s = await c?.shell(pty: SSHPtyConfig(type: 'ansi-m')/*, environment: {
      "LS_COLORS": "di=30:ln=30:so=30:pi=30:ex=30:bd=30:cd=30:su=30:sg=30:tw=30:ow=30",
      "GREP_COLORS": "di=30:ln=30:so=30:pi=30:ex=30:bd=30:cd=30:su=30:sg=30:tw=30:ow=30",
    }*/);
    return s;
  }

  Future<SSHSession?> excute(String cmd) async {
    final c = await client;
    return await c?.execute(cmd, pty: SSHPtyConfig(type: 'ansi-m'));
  }

  List<String> toList() {
    return [ip, user, pswd, port.toRadixString(10)];
  }
}

class SshCfgEvent {
  static final int eventIP = 1;
  static final int eventCluster = 2;
  int event;
  String name;
  SshCfgEvent(this.event, this.name);
}

Map<String, String> customCmd = {};

class IsolateMessage {
  static const int TypeDbg = 0;
  static const int TypeInit = 1;
  static const int TypeUnInit = 2;
  static const int TypeSshStart = 3;
  static const int TypeSshStop = 4;
  static const int TypeDead = 5;
  static const int _TypeCallbackBegin = 100;
  static const int TypeLines = 100;
  static const int TypeSshComplete = 101;

  final int type; 
  final Object data;
  IsolateMessage(this.type, this.data);
  bool get isCallback => type >= _TypeCallbackBegin;
}

typedef OneArgFunction<T,K> = T Function(K arg);
typedef TwoArgFunction<T,K,J> = T Function(K arg1, J arg2);

class IsolateSshWorker {
  bool _dead = false;
  final ReceivePort _hostPort = ReceivePort();
  final ReceivePort _hostErrPort = ReceivePort();
  Completer ? _busy;
  late SendPort _sendPort;
  Isolate ? _isolate;
  Function(IsolateMessage) ? _onMessage;
  bool get dead => _dead;

  Future<bool> init() async {
    if (_isolate!=null)  return true;
    _isolate = await Isolate.spawn(_workerLoop, _hostPort.sendPort);
    if (_isolate == null)   return false;
    final waitInit = Completer();
    _hostPort.listen((message) {
      final msg = message as IsolateMessage;
      if ( msg.type == IsolateMessage.TypeInit) {
        _sendPort = msg.data as SendPort;
        waitInit.complete();
      } else if ( msg.type == IsolateMessage.TypeDead) {
        _onDead();
      } 
      if ( msg.isCallback ) {
        if (_onMessage!=null) _onMessage!(message);
      }
    });
    _hostErrPort.listen((message) {
      _onDead();
    });
    _isolate?.addErrorListener(_hostErrPort.sendPort);
    await waitInit.future;
    return true;
  }

  void _onDead() {
    print("Isolate working dead========== ${_sendPort.hashCode}");
    if (_onMessage != null) {
      _onMessage!(IsolateMessage(IsolateMessage.TypeSshComplete, CmdResult(false, "ssh over")));
      _onMessage = null;
    }
    if (_busy != null) {
      _busy?.complete();
      _busy = null;
    }
    _dead = true;
    _isolate?.kill();
    _hostPort.close();
    _hostErrPort.close();
    _isolate = null;
  }

  Future<void> start(SshCfg ssh, String cmd, void Function(List<String> lines) onLines, void Function(CmdResult res) onCompelet) async {
    print("Isolate[${_sendPort.hashCode}] wait +, $_busy, cmd=$cmd");
    await _busy?.future;
    print("Isolate[${_sendPort.hashCode}] wait -, $_busy");
    _busy = Completer();
    _onMessage = (IsolateMessage msg) {
      if (msg.type == IsolateMessage.TypeLines) {
        onLines(msg.data as List<String>);
      } else if (msg.type == IsolateMessage.TypeSshComplete) {
        _onMessage = null;
        onCompelet(msg.data as CmdResult);
        _busy?.complete();
        _busy = null;
      }
    };
    print("Isolate start, ${_sendPort.hashCode}");
    _sendPort.send(IsolateMessage(IsolateMessage.TypeSshStart, [ssh, cmd]));
    await _busy?.future; 
    print("Isolate stop");
  }

  Future<void> stop() async {
    _sendPort.send(IsolateMessage(IsolateMessage.TypeSshStop, 0));
    await _busy?.future;
  }

  static void _workerLoop(SendPort hostPort) {
    final workerPort = ReceivePort();
    print("Isolate working enable: ${workerPort.sendPort.hashCode}");
    hostPort.send(IsolateMessage(IsolateMessage.TypeInit, workerPort.sendPort));
    AbortCallback ? abort;
    Completer ? stopWait;
    workerPort.listen((message) {
      final msg = message as IsolateMessage;
      print("worker[${workerPort.sendPort.hashCode}]: ${msg.type}, ${msg.data}");
      if (msg.type == IsolateMessage.TypeSshStart) {
        print("Isolate working....");
        final data = msg.data as List<Object>;
        final ssh = data[0] as SshCfg;
        final cmd = data[1] as String;
        abort = shellAsync(ssh, cmd, (lines) { 
          hostPort.send(IsolateMessage(IsolateMessage.TypeLines, lines));
        }, (res) { 
          abort = null;
          stopWait?.complete();
          hostPort.send(IsolateMessage(IsolateMessage.TypeSshComplete, res));
          print("Isolate working end!!");
        });
        print("worker[${workerPort.sendPort.hashCode}]: abort=${abort.hashCode}");
      } else if (msg.type == IsolateMessage.TypeSshStop) {
        if (abort!=null) {
          print("worker[${workerPort.sendPort.hashCode}]: abort=${abort.hashCode} stop+!!");
          stopWait = Completer();
          abort!();
          stopWait?.future.timeout(Duration(milliseconds: 500)).onError((error, stackTrace) {
            print("worker[${workerPort.sendPort.hashCode}]: abort=${abort.hashCode} stop-err=$error");
          }).whenComplete(() {
            stopWait = null;
            print("worker[${workerPort.sendPort.hashCode}]: abort=${abort.hashCode} stop-!!");
            if (abort != null) {
              print("Isolate working dead++!!");
              hostPort.send(IsolateMessage(IsolateMessage.TypeSshComplete, CmdResult(false, "ssh dead")));
              hostPort.send(IsolateMessage(IsolateMessage.TypeDead, false));
              print("Isolate working dead--!!");
            }
          });
        } 
      }
    });
  }

  static AbortCallback? shellAsync(
      SshCfg ssh,
      String cmd,
      void Function(List<String> lines) onLines,
      void Function(CmdResult res) onCompelet) {
    SSHSession? rs;
    StreamSubscription<Uint8List>? rout;
    StreamSubscription<Uint8List>? rerr;
    final abort = () async {
      if (rs == null)   return;
      print("Abort + =>$cmd,$rout,$rerr,$rs");
      try {
        print("kill Abort");
        rs!.kill(SSHSignal.ABRT);
      } catch (e) {
        print("kill-error: $e");
      }
      try {
        print("close");
        rs!.close();
      } catch (e) {
      } finally {
        await rs!.done.timeout(Duration(milliseconds: 100));
        rs = null;
      }
      try {
        print("rout cancel");
        rout?.cancel();
      } catch (err) {
      } finally {
        rout = null;
      }
      try {
        print("rerr cancel");
        rerr?.cancel();
      } catch (err) {
      } finally {
        rerr = null;
      }
      print("Abort - =>$cmd");
    };
    final session = ssh.excute(cmd).catchError((e) {
      return null;
    }).then((value) {
      rs = value;
      return rs;
    });
    session.then((session) async {
      if (session == null) {
        onCompelet(CmdResult(false, "ssh login failed: ${ssh.ip}"));
        return;
      }
      final result = BytesBuilder(copy: false);
      final stdoutDone = Completer<void>();
      final stderrDone = Completer<void>();
      final makeLines = (List<int> bytes) {
        int start = 0;
        List<String> lines = [];
        for (int i = 0; i < bytes.length; i++) {
          if (bytes[i] == 10) {
            final bline = bytes.sublist(start, i);
            late String res;
            if (result.length > 0) {
              result.add(bline);
              res = utf8.decode(result.takeBytes(), allowMalformed: true);
            } else {
              res = utf8.decode(bline, allowMalformed: true);
            }
            start = i + 1;
            res = res.replaceAll("\t", "  ");
            lines.add(res);
          }
        }
        if (start < bytes.length) {
          result.add(bytes.sublist(start));
        }
        return lines;
      };
      final add = (List<int> bytes) {
        final lines = makeLines(bytes);
        if (lines.isNotEmpty) onLines(lines);
      };

      rout = session.stdout
          .listen(add, onDone: stdoutDone.complete, cancelOnError: true);
      rerr = session.stderr
          .listen(add, onDone: stderrDone.complete, cancelOnError: true);

      await stdoutDone.future;
      await stderrDone.future;
      if (result.length > 0) {
        add(result.takeBytes());
      }
    }).catchError((err) {
      print("CatchError=>$cmd");
    }).whenComplete(() {
      print("Complete=>$cmd");
      abort();
      onCompelet(CmdResult(true, ""));
    });
    return abort;
  }
}

class SshService extends GetxService {
  final sockets = <String, SSHSocket>{};
  final sshcfg = <String, SshCfg>{};
  final clusters = <String, List<String>>{};
  final workers = <IsolateSshWorker?>[];
  final EventBus eventBus = EventBus(sync: false);
  late final notifyIpEvent = Tools.debounce((ip) {
    SpUtils.put(
        "SSH_CFG",
        sshcfg.map((key, value) =>
            MapEntry<String, List<String>>(key, value.toList())));
    eventBus.fire(SshCfgEvent(SshCfgEvent.eventIP, ip));
  }, Duration(milliseconds: 100));
  late final notifyClusterEvent = Tools.debounce((cluster) {
    SpUtils.put("CLUSTER_CFG", clusters);
    eventBus.fire(SshCfgEvent(SshCfgEvent.eventCluster, cluster));
  }, Duration(milliseconds: 100));


  Future<SshService> init() async {
    final ov = SpUtils.get("SSH_CFG", initValue: <String, dynamic>{});
    sshcfg.clear();
    ov.forEach((key, value) {
      sshcfg[key] = SshCfg.fromList(value);
    });
    final oc = SpUtils.get("CLUSTER_CFG", initValue: <String, dynamic>{});
    clusters.clear();
    oc.forEach((key, value) {
      List<String> ipList = [];
      for (int i = 0; i < value.length; i++) {
        ipList.add(value[i]);
      }
      clusters[key] = ipList;
    });
    loadCustomCmd();
    makeWorkers(5);
    return this;
  }

  Iterable<String> get allIpListNotInCluster {
    return sshcfg.keys.where((value) => !clusters.values
        .any((element) => element.any((element) => element == value)));
  }

  Iterable<String> get allIpList {
    return sshcfg.keys;
  }

  Iterable<String> get allClusterList {
    return clusters.keys;
  }

  void makeWorkers(int count) async {
    List waiters = [];
    List result = [];
    for (int i=0; i<count; i++) {
      final worker = IsolateSshWorker();
      waiters.add(worker.init());
      result.add(worker);
    }
    for (int i=0; i<count; i++) {
      if (await waiters[i]) {
        workers.add(result[i]);
      }
    }
  }

  Future<IsolateSshWorker?> reqWorker() async {
    IsolateSshWorker ? worker;
    for (int i=0; i<workers.length; i++) {
      worker = workers[i];
      if (worker != null) {
        workers[i] = null;
        return worker;
      }
    }
    worker = IsolateSshWorker();
    if (!await worker.init()) return null;
    workers.add(worker);
    return worker;
  }

  void fireWorker(IsolateSshWorker worker) {
    print("fireworker ==== ${worker.hashCode} ${worker.dead}");
    if (worker.dead) {
      final worker = IsolateSshWorker();
      worker.init().then((ok){
        if (ok) workers.add(worker);
      });
      return ;
    }
    int pos = -1;
    for (int i=0; i<workers.length; i++) {
      if (pos < 0 && workers[i] == null) {
        pos = i;
      }
      if (workers[i] == worker) {
        print("double fired");
        return ;
      }
    }
    if (pos >= 0) {
      workers[pos] = worker;
    }
  }

  void updateSshCfg(String ip, String user, String pswd, int port) {
    sshcfg[ip] = SshCfg(ip, user, pswd, port);
    notifyIpEvent(ip);
  }

  void deleteIp(String ip) {
    sshcfg.remove(ip);
    notifyIpEvent(ip);
  }

  void updateCluster(String cluster, List<String> ipList) {
    clusters[cluster] = ipList;
    notifyClusterEvent(cluster);
  }

  void deleteCluster(String cluster) {
    if (!clusters.containsKey(cluster)) return;
    final removeIpList = clusters[cluster];
    sshcfg.removeWhere((key, value) => removeIpList!.contains(key));
    clusters.remove(cluster);
    notifyClusterEvent(cluster);
  }

  Future<String> makeShell(
      String ip, int port, String cmd, String user, String pswd) async {
    final sshcfg = SshCfg(ip, user, pswd, port);
    final client = await sshcfg.client.onError((error, stackTrace) {
      Get.snackbar("error", "ssh login failed: $ip, error: $error");
    });
    if (client == null) return "";
    final res = await client.run(cmd);
    client.close();
    return utf8.decode(res);
  }

  static AbortCallback? shellAsync(
      SshCfg ssh,
      String cmd,
      void Function(List<String> lines) onLines,
      void Function(CmdResult res) onCompelet) {
    SSHSession? rs;
    StreamSubscription<Uint8List>? rout;
    StreamSubscription<Uint8List>? rerr;
    final abort = () {
      try {
        rout?.cancel();
      } catch (err) {
      } finally {
        rout = null;
      }
      try {
        rerr?.cancel();
      } catch (err) {
      } finally {
        rerr = null;
      }
      try {
        rs?.close();
      } catch (e) {
      } finally {
        rs = null;
      }
    };
    final session = ssh.excute(cmd).catchError((e) {
      return null;
    }).then((value) {
      rs = value;
      return rs;
    });
    session.then((session) async {
      if (session == null) {
        onCompelet(CmdResult(false, "ssh login failed: ${ssh.ip}"));
        return;
      }
      final result = BytesBuilder(copy: false);
      final stdoutDone = Completer<void>();
      final stderrDone = Completer<void>();
      final makeLines = (List<int> bytes) {
        int start = 0;
        List<String> lines = [];
        for (int i = 0; i < bytes.length; i++) {
          if (bytes[i] == 10) {
            final bline = bytes.sublist(start, i);
            late String res;
            if (result.length > 0) {
              result.add(bline);
              res = utf8.decode(result.takeBytes(), allowMalformed: true);
            } else {
              res = utf8.decode(bline, allowMalformed: true);
            }
            start = i + 1;
            res = res.replaceAll("\t", "  ");
            lines.add(res);
          }
        }
        if (start < bytes.length) {
          result.add(bytes.sublist(start));
        }
        return lines;
      };
      final add = (List<int> bytes) {
        final lines = makeLines(bytes);
        if (lines.isNotEmpty) onLines(lines);
      };

      rout = session.stdout
          .listen(add, onDone: stdoutDone.complete, cancelOnError: true);
      rerr = session.stderr
          .listen(add, onDone: stderrDone.complete, cancelOnError: true);

      await stdoutDone.future;
      await stderrDone.future;
      if (result.length > 0) {
        add(result.takeBytes());
      }
    }).catchError((err) {
      abort();
    }).whenComplete(() {
      abort();
      onCompelet(CmdResult(true, ""));
    });
    return abort;
  }

  Future<AbortCallback?> shellAsync2(
      String ip,
      String cmd,
      void Function(List<String> lines) onLines,
      void Function(CmdResult res) onCompelet) async {
    final ssh = sshcfg[ip];
    if (ssh==null) {
      onCompelet(CmdResult(false, "cfg not exist"));
      return null;
    }
    final ret = await reqWorker().then((worker) {
      if (worker == null) {
        onCompelet(CmdResult(false, "system busy"));
        return null;
      }
      printInfo(info:"========= worker: ${worker.hashCode}");
      final release = (){
        worker?.stop().then((msg){
          if (worker!=null) {
            fireWorker(worker!);
            worker = null;
          }
        }, onError: (err){
          printInfo(info:"========= worker: ${worker.hashCode} Abort Error: $err");
        });
      };
      worker?.start(ssh, cmd, onLines, onCompelet).then((msg) {
        release();
      });
      return release;
    });
    return ret;
  }


  void loadCustomCmd() {
    final file = File("custom_cmd.yaml");
    file.readAsString().then((value) {
      final cfg = loadYaml(value);
      if (cfg.runtimeType == YamlMap) {
        final map = cfg as YamlMap;
        customCmd.clear();
        map.forEach((key, value) {
          if (key.runtimeType != String)  return;
          if (key=="CONST")  return;
          if (value.runtimeType == String) {
            customCmd[key] = value;
          }
          if (value.runtimeType == YamlList) {
            customCmd[key] = (value as YamlList).join("\n");
          }
        });
      }
    });
  }
}
